package com.atguigu.day07;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

// select * from A inner join B on A.key = B.key;
public class Example3 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        var leftStream = env
                .fromElements(
                        Tuple2.of("key-1", "left-1"),
                        Tuple2.of("key-1", "left-2"),
                        Tuple2.of("key-2", "left-1")
                );

        var rightStream = env
                .fromElements(
                        Tuple2.of("key-1", "right-1"),
                        Tuple2.of("key-1", "right-2"),
                        Tuple2.of("key-2", "right-1")
                );

        leftStream.keyBy(r -> r.f0)
                .connect(rightStream.keyBy(r -> r.f0))
                .process(new InnerJoin())
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class InnerJoin extends CoProcessFunction<Tuple2<String, String>, Tuple2<String, String>, String> {
        private ListState<Tuple2<String, String>> leftHistory;
        private ListState<Tuple2<String, String>> rightHistory;

        @Override
        public void open(Configuration parameters) throws Exception {
            leftHistory = getRuntimeContext().getListState(
                    new ListStateDescriptor<>(
                            "left-state",
                            Types.TUPLE(Types.STRING, Types.STRING)
                    )
            );
            rightHistory = getRuntimeContext().getListState(
                    new ListStateDescriptor<>(
                            "right-state",
                            Types.TUPLE(Types.STRING, Types.STRING)
                    )
            );
        }

        @Override
        public void processElement1(Tuple2<String, String> left, Context context, Collector<String> collector) throws Exception {
            leftHistory.add(left);

            for (var right : rightHistory.get()) {
                collector.collect(left + " ===> " + right);
            }
        }

        @Override
        public void processElement2(Tuple2<String, String> right, Context context, Collector<String> collector) throws Exception {
            rightHistory.add(right);

            for (var left : leftHistory.get()) {
                collector.collect(left + " ==> " + right);
            }
        }
    }
}
